package com.yomob.kylin.service;

import com.clearspring.analytics.util.Lists;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.UpdateManyModel;
import com.mongodb.client.model.UpdateOptions;
import com.yomob.kylin.client.MongoDBClient;
import com.yomob.kylin.facility.Consts;
import com.yomob.kylin.support.IpUtil;
import com.yomob.kylin.support.OptionSupport;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.storage.StorageLevel;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.URISyntaxException;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

import static com.mongodb.client.model.Filters.eq;
import static org.apache.spark.sql.functions.col;

/**
 * collect device(user) info, one device represent one user.
 * collect info contain static and dynamic data, such as : device unique id,device brand,net type,position,active time...
 */
public class NewUserService extends SimpleService{
    private static final Logger LOGGER = LoggerFactory.getLogger(NewUserService.class);
    private static final String COLLECTION_NAME = "daily_user_info";
    private MongoDBClient mongoDBClient;
    private IpUtil ipUtil;
    //for new user counter may no contain udid and system,just used active counter data
    private static final Object[] COUNTERIDS = {"active"};
    public NewUserService(MongoDBClient mongoDBClient) throws IOException, URISyntaxException {
        this.mongoDBClient = mongoDBClient;
        ipUtil = new IpUtil();
    }
    /**
     * Running Entry for new user service
     * @param input,spark data set during the start-day and end-day
     * @throws Exception
     */
    @Override
    public void execute(Dataset<Row> input) throws Exception {
        Dataset<Row> dataset = input.filter(col(Consts.Fields.COUNTERID).isin("active")).select(
                Consts.Fields.UDID,
                Consts.Fields.EXT$SYSTEM,
                Consts.Fields.USERID,
                Consts.Fields.APPID,
                Consts.Fields.PLATFORM,
                //active data
                Consts.Fields.DATE,
                Consts.Fields.HOUR,
                Consts.Fields.EXT$NET,

                Consts.Fields.COUNTRY,
                Consts.Fields.EXT$IP
        ).distinct();
        dataset.persist(StorageLevel.fromString(OptionSupport.getStorageLevel().name()));
        List<Row> rowList = dataset.collectAsList();
        List<UpdateManyModel<Document>> models = Lists.newArrayList();
        models.addAll(rowList.stream().map(row -> this.convert(row)).collect(Collectors.toList()));
        mongoDBClient.saveToMongoBulkByBatch(models, COLLECTION_NAME);
        dataset.unpersist();
    }

    /**
     * return this service deal counters type
     * @return String array
     */
    @Override
    public Object[] getCounterids() {
        return COUNTERIDS;
    }

    /**
     * Private method,used to convert spark row data to mongo update command set,contain $set and $addToSet.
     * @param row
     * @return mongodb UpdateManyModel<Document> for batch update
     */
    private UpdateManyModel<Document> convert(Row row) {
        List<Bson> matches = Lists.newArrayList();
        Bson eq = eq(Consts.UserInfoBased.DEVICE_ID, row.getAs(Consts.Fields.UDID));
        matches.add(eq);
        Bson filter = Filters.and(matches);
        Document set = new Document();
        Document update = new Document();
        update.put(Consts.UserInfoBehavior.APP_ID,row.getAs(Consts.Fields.APPID));
        String sys = row.getAs(1);
        if(sys != null && sys.toString().trim().length() > 0) {
            String[] systemInfo = sys.split("_");
            update.put(Consts.UserInfoBased.DEVICE_BRAND, systemInfo[0].trim());
        }
        String ip = row.getAs("ip");
        String[] cityInfo = ipUtil.getCityInfo(ip);
        if(cityInfo != null){
            update.put(Consts.UserInfoBased.CITY, cityInfo[0]);
            update.put(Consts.UserInfoBased.PROVINCE, cityInfo[1]);
        }
        update.put(Consts.Fields.PLATFORM,row.getAs(Consts.Fields.PLATFORM));
        update.put(Consts.UserInfoBased.COUNTRY, row.getAs(Consts.Fields.COUNTRY));
        update.put(Consts.UserInfoBased.IP, ip);
        update.put(Consts.UserInfoBehavior.ACTIVE_TIME,row.getAs(Consts.Fields.DATE));
        update.put(Consts.UserInfoBehavior.DAY_PHASE, getDayPhase(row.getAs(Consts.Fields.HOUR)));
        update.put(Consts.UserInfoBehavior.DAY_TYPE, getDayType(row.getAs(Consts.Fields.DATE)));
        update.put(Consts.UserInfoBehavior.DEVICE_NETWORK, row.getAs("net"));
        set.append(Consts.MongoCommand.SET,update);
        Document uid = new Document();
        uid.put(Consts.UserInfoBased.UNIONIDS, row.getAs(Consts.Fields.APPID) + "-" + row.getAs(Consts.Fields.USERID));//appID-userID
        set.append(Consts.MongoCommand.ADD_TO_SET,uid);
        UpdateOptions updateOptions = new UpdateOptions().upsert(true);
        return new UpdateManyModel<>(filter, set, updateOptions);
    }

    /**
     * @param date yyyy-MM-dd
     * @return date type (weekend or workday)
     */
    public static String getDayType(String date) {
        DateFormat format1 = new SimpleDateFormat("yyyy-MM-dd");
        Date bdate = null;
        try {
            bdate = format1.parse(date);
        } catch (ParseException e) {
            e.printStackTrace();
            LOGGER.warn("Get an invalid date '{}',the correct format is yyyy-MM-dd .", date);
            //todo
            bdate = new Date();
        }
        Calendar cal = Calendar.getInstance();
        cal.setTime(bdate);
        if (cal.get(Calendar.DAY_OF_WEEK) == Calendar.SATURDAY || cal.get(Calendar.DAY_OF_WEEK) == Calendar.SUNDAY) {
            return "weekend";
        } else {
            return "workday";
        }
    }

    /**
     * private method,get the phase of the day
     * @param dateHour yyyy-MM-ddThh,this data-value come from row counter's UTC dateTime.
     * @return am,noon,pm,night,deepNight
     */
    private static String getDayPhase(String dateHour) {
        String a = "";
        String hour = dateHour.substring(dateHour.length() - 2);
        int h = 0;
        if (hour.startsWith("0")) {
            h = Integer.valueOf(hour.substring(1));
        } else {
            h = Integer.valueOf(hour);
        }
        if (h > 6 && h <= 11) {
            a = "am";
        } else if (h > 11 && h < 14) {
            a = "noon";
        } else if (h >= 14 && h <= 18) {
            a = "pm";
        } else if (h > 18 && h < 24) {
            a = "night";
        } else {
            a = "deepNight";
        }
        return a;
    }
}